package cn.doitedu.sql;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/4
 * @Desc: 学大数据，上多易教育
 *   左表数据，去关联右表一段时间范围内的数据
 **/
public class _13_IntervalJoin {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(1000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        tenv.getConfig().set("table.exec.source.idle-timeout","1s");
        tenv.getConfig().set("table.exec.state.ttl","1 hour");


        // {"uid":1,"event_id":"face_recg","action_time":10000}
        tenv.executeSql(
                " create table face_log_kfk(                                    "+
                        "       uid int,                                        "+
                        " 	    event_id string,                                "+
                        " 	    action_time bigint,                             "+
                        " 	    pt as proctime(),                               "+
                        " 	    rt as to_timestamp_ltz(action_time,3),          "+
                        " 	    watermark for rt as rt                          "+
                        "  ) with (                                             "+
                        "       'connector' = 'kafka',                           "+
                        "       'topic' = 'face-log',                            "+
                        "       'properties.bootstrap.servers' = 'doitedu:9092', "+
                        "       'properties.group.id' = 'g003',                  "+
                        "       'scan.startup.mode' = 'latest-offset',           "+
                        "       'value.format' = 'json',                         "+
                        "       'value.fields-include' = 'EXCEPT_KEY'            "+
                        " )                                                      "
        );

        // {"uid":1,"face_status":1,"ts":11000}
        tenv.executeSql(
                " create table face_res_kfk(                                    "+
                        "   uid int,                                            "+
                        " 	face_status int,                                    "+
                        " 	ts bigint,                                          "+
                        " 	pt as proctime(),                                   "+
                        " 	rt as to_timestamp_ltz(ts,3),                       "+
                        " 	watermark for rt as rt                              "+
                        "  ) with (                                             "+
                        "    'connector' = 'kafka',                             "+
                        "    'topic' = 'face-res',                            "+
                        "    'properties.bootstrap.servers' = 'doitedu:9092',   "+
                        "    'properties.group.id' = 'g003',                    "+
                        "    'scan.startup.mode' = 'latest-offset',             "+
                        "    'value.format' = 'json',                           "+
                        "    'value.fields-include' = 'EXCEPT_KEY'              "+
                        " )                                                     "
        );

        tenv.executeSql(
                        " select                                   "+
                        "  a.uid                                   "+
                        " ,a.event_id                              "+
                        " ,a.action_time                           "+
                        " ,b.face_status                           "+
                        " ,b.ts                                    "+
                        "                                          "+
                        " from face_log_kfk a                      "+
                        " join face_res_kfk b                      "+
                        " on a.uid=b.uid                           "+
                        " and a.rt <= b.rt + interval '10' second  "+
                        " and a.rt >= b.rt - interval '10' second  "
        ).print();



    }
}
